package com.seventeen.monster.mq.proxy.server.proxy.consumer;

import com.seventeen.monster.common.constant.InstanceTypeConstants;
import com.seventeen.monster.mq.proxy.server.proxy.thread.ConsumerPool;
import com.seventeen.monster.mq.proxy.server.proxy.ext.BaseExtMqConsumer;
import com.seventeen.monster.mq.proxy.server.proxy.ext.DefaultExtMqPushConsumer;
import com.seventeen.monster.mq.proxy.server.proxy.ext.NewExtMqPushConsumer;
import com.seventeen.monster.mq.proxy.domain.MqConsumerProxy;
import com.seventeen.monster.mq.proxy.domain.MqInstance;
import com.seventeen.monster.mq.proxy.service.IMqConsumerProxyService;
import com.seventeen.monster.mq.proxy.service.IMqInstanceService;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 消费者代理工厂
 *
 * @author seventeen
 */
@Component
public class ConsumerFactory implements InitializingBean, DisposableBean {

    public static final Logger LOGGER = LoggerFactory.getLogger(ConsumerFactory.class);

    private final ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1,
            new BasicThreadFactory.Builder().namingPattern("consumer-factory-schedule-pool-%d").daemon(true).build());
    private ConsumerPool consumerPool;

    @Autowired
    private IMqConsumerProxyService mqConsumerProxyService;
    @Autowired
    private IMqInstanceService mqInstanceService;

    public ConsumerPool getConsumerPool() {
        return consumerPool;
    }

    @Override
    public void destroy() throws Exception {
        LOGGER.info("consumerFactory destroy!");
        consumerPool.toStop();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            LOGGER.warn("", e);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        consumerPool = new ConsumerPool();
        consumerPool.setDaemon(true);
        consumerPool.start();
        this.scheduledExecutorService.scheduleAtFixedRate(() ->
        {
            try {
                mqConsumerProxyService.list().forEach(this::addConsumerProxy);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
        }, 1000 * 2, 1000 * 60, TimeUnit.MILLISECONDS);
    }

    public void addConsumerProxy(MqConsumerProxy mqConsumerProxy) {
        Long instanceId = mqConsumerProxy.getInstanceId();
        MqInstance mqInstance = mqInstanceService.selectById(instanceId);
        BaseExtMqConsumer mqConsumer = getExtMqProducerByInstanceType(mqInstance.getInstanceType());
        mqConsumer.setId(mqConsumerProxy.getId());
        mqConsumer.setAppUrl(mqConsumerProxy.getAppUrl());
        mqConsumer.setConsumerProxyGroup(mqConsumerProxy.getConsumerProxyGroup());
        mqConsumer.setTopic(mqConsumerProxy.getTopic());
        mqConsumer.setTags(mqConsumerProxy.getTags());
        mqConsumer.setVersion(mqConsumerProxy.getVersion());
        mqConsumer.setType(mqConsumerProxy.getMsgType());

        mqConsumer.setAccessKey(mqInstance.getAccessKey());
        mqConsumer.setSecretKey(mqInstance.getSecretKey());
        mqConsumer.setEndpoints(mqInstance.getEndpoints());
        consumerPool.pushQueue(mqConsumer);
    }

    private BaseExtMqConsumer getExtMqProducerByInstanceType(String instanceType) {

        BaseExtMqConsumer mqConsumer;
        switch (instanceType) {
            case InstanceTypeConstants.V4_X:
            case InstanceTypeConstants.V_ALI_CLOUD_4_X:
                mqConsumer = new DefaultExtMqPushConsumer();
                break;
            case InstanceTypeConstants.V5_X:
            case InstanceTypeConstants.V_ALI_CLOUD_5_X:
                mqConsumer = new NewExtMqPushConsumer();
                break;
            default:
                mqConsumer = null;
        }
        return mqConsumer;
    }
}
